package com.github.ghsea.retry;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableList;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import javax.annotation.PreDestroy;
import javax.sql.DataSource;
import javax.validation.ConstraintViolation;
import javax.validation.Validation;
import javax.validation.Validator;
import javax.validation.ValidatorFactory;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.joda.time.Period;
import org.killbill.billing.util.queue.QueueRetryException;
import org.killbill.clock.DefaultClock;
import org.killbill.commons.jdbi.notification.DatabaseTransactionNotificationApi;
import org.killbill.commons.jdbi.transaction.NotificationTransactionHandler;
import org.killbill.notificationq.DefaultNotificationQueueService;
import org.killbill.notificationq.api.NotificationEvent;
import org.killbill.notificationq.api.NotificationQueue;
import org.killbill.notificationq.api.NotificationQueueService;
import org.killbill.queue.InTransaction;
import org.killbill.queue.retry.RetryableHandler;
import org.killbill.queue.retry.RetryableService;
import org.skife.jdbi.v2.DBI;

/**
 * @author guhai
 */

@Slf4j
public abstract class AbstractRetryableAction<T> {

    private NotificationQueueService queueService;

    private NotificationQueue queueWithRetry;

    private RetryableNotificationQueueService retryableQueueService;

    private DataSource dataSource;

    private DBI dbi;

    private final int DELAY_SECOND = 1;

    private final static String TABLE_NOTIFICATION = "notifications";

    private final static int errorCountBeforeRetrySchedule = 2;

    private final String SQL_QUERY_NOTIFICATION = "select error_count,class_name from %s where search_key2= %d and search_key1= %d order by record_id desc ";

    private static ValidatorFactory factory = Validation.buildDefaultValidatorFactory();

    public void execute(RetryableEvent<T> event) {
        validate(event);

        try {
            doExecute(event.getActionParamBean());
        } catch (ActionRetryableException e) {
            //加入重试队列
            log.error("error for event:{}", event.toString(), e);
            recordEvent(event);
        }
    }

    private void validate(RetryableEvent<T> event) {
        Validator validator = factory.getValidator();
        Set<ConstraintViolation<T>> constraintViolations = validator.validate(event.getActionParamBean());
        if (constraintViolations != null && constraintViolations.size() > 0) {
            List<String> messages = new ArrayList<>(constraintViolations.size());
            constraintViolations.forEach(e -> messages.add(e.getPropertyPath().toString() + e.getMessage()));
            throw new ValidationException(messages);
        }
    }


    protected abstract void doExecute(T param) throws ActionRetryableException;

    /**
     * 获取重试action的类名
     *
     * @return
     */
    protected abstract Class getActionClass();

    protected abstract String getServiceName();

    /**
     * 重试时机
     *
     * @return 重试的时间间隔
     */
    protected List<Period> getRetrySchedule() {
        return ImmutableList.of(
                Period.seconds(1),
                Period.seconds(10),
                Period.minutes(5));
    }

    public AbstractRetryableAction(DataSource dataSource) {
        Objects.nonNull(dataSource);
        this.dataSource = dataSource;
        initRetryableQueueService();
    }

    public void initRetryableQueueService() {
        dbi = new DBI(dataSource);
        InTransaction.setupDBI(dbi);
        dbi.setTransactionHandler(new NotificationTransactionHandler(new DatabaseTransactionNotificationApi()));

        /**
         * @see PersistentBusConfi,org.killbill.notificationq.api.NotificationQueueConfig
         */
        Properties properties = new Properties();
        properties.put("org.killbill.persistent.bus.main.sleep", 100);
        properties.put("org.killbill.persistent.bus.main.nbThreads", 1);
        properties.put("org.killbill.notificationq.main.sleep", 100);
        properties.put("org.killbill.notificationq.main.max.failure.retry", 10);

        queueService = new DefaultNotificationQueueService(dataSource, properties);
        retryableQueueService = new RetryableNotificationQueueService(queueService);

        OrderNotificationQueueHandler handlerDelegate = new OrderNotificationQueueHandler();
        final NotificationQueueService.NotificationQueueHandler retryableHandler = new RetryableHandler(new DefaultClock(), retryableQueueService, handlerDelegate);
        try {
            queueWithRetry = queueService.createNotificationQueue(getServiceName(),
                    "AbstractRetryableAction", retryableHandler);
        } catch (NotificationQueueService.NotificationQueueAlreadyExists ex) {
            throw new RuntimeException(ex);
        }
        queueWithRetry.startQueue();

        retryableQueueService.initialize(queueWithRetry.getQueueName(), handlerDelegate);
        retryableQueueService.start();
    }

    @PreDestroy
    public void destroy() {
        queueWithRetry.stopQueue();
        try {
            retryableQueueService.stop();
        } catch (NotificationQueueService.NoSuchNotificationQueue ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    private void recordEvent(RetryableEvent event) {
        dbi.inTransaction((conn, status) -> {
            long key2 = System.currentTimeMillis();
            final UUID TOKEN_ID = UUID.randomUUID();
            log.info("record event,uuid:{}", TOKEN_ID);
            DateTime readyTime = new DateTime().plusSeconds(DELAY_SECOND);

            RetryableEventWrapper eventWrapper = new RetryableEventWrapper();
            eventWrapper.setActionClassName(getActionClass().getName()).setEvent(event);
            event.setKey2(key2);

            queueWithRetry.recordFutureNotificationFromTransaction(conn.getConnection(), readyTime, eventWrapper, TOKEN_ID, event.getKey(), event.getKey2());

            return null;
        });
    }


    private class OrderNotificationQueueHandler implements NotificationQueueService.NotificationQueueHandler {

        @Override
        public void handleReadyNotification(NotificationEvent eventJson, DateTime eventDateTime, UUID userToken, Long searchKey1, Long searchKey2) {
            RetryableEventWrapper event = JSON.parseObject(eventJson.toString(), RetryableEventWrapper.class);
            try {
                AbstractRetryableAction action = DefaultActionFactory.getAction(Class.forName(event.getActionClassName()));

                ParameterizedType paramTypes = (ParameterizedType) action.getClass().getGenericSuperclass();
                Class paramType = (Class) paramTypes.getActualTypeArguments()[0];
                Object actionParamBean = JSON.parseObject(JSON.toJSONString(event.getEvent().getActionParamBean()), paramType);

                action.doExecute(actionParamBean);

            } catch (ActionRetryableException retryEx) {
                retryEx.printStackTrace();
                RetryableEvent eventBean = event.getEvent();
                if (isRetryNotificationEvent(eventBean.getKey2(), eventBean.getKey())) {
                    throw new QueueRetryException(retryEx, getRetrySchedule());
                } else {
                    throw retryEx;
                }
            } catch (QueueRetryException ex) {
                throw ex;
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            }
        }
    }

    private boolean isRetryNotificationEvent(long key2, long key1) {
        Notification record = dbi.withHandle(handle -> {
            String sql = String.format(SQL_QUERY_NOTIFICATION, TABLE_NOTIFICATION, key2, key1);
            return handle.createQuery(sql).map((idx, rs, ctx) -> new Notification(rs.getInt("error_count"), rs.getString("class_name"))).first();
        });

        boolean isRetryNotificationEvent = "org.killbill.queue.retry.RetryNotificationEvent".equals(record.getClassName());
        return isRetryNotificationEvent || record.getErrorCount() == errorCountBeforeRetrySchedule;
    }


    private static final class RetryableNotificationQueueService extends RetryableService {

        RetryableNotificationQueueService(final NotificationQueueService notificationQueueService) {
            super(notificationQueueService);
        }
    }

    private final class Notification {
        private String className;
        private int errorCount;

        Notification(int errorCount, String className) {
            this.errorCount = errorCount;
            this.className = className;
        }

        String getClassName() {
            return className;
        }

        Notification setClassName(String className) {
            this.className = className;
            return this;
        }

        int getErrorCount() {
            return errorCount;
        }

        Notification setErrorCount(int errorCount) {
            this.errorCount = errorCount;
            return this;
        }
    }

}
